package com.ibm.cps.message;

import java.util.HashMap;
import java.util.Set;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoSerializable;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.gson.Gson;
import com.ibm.cps.message.MessageFactory.MessageType;
import com.ibm.cps.processors.deletion.NullDeletion;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

@Deprecated
public class SchemaMetaMessage extends ProcessorMetaMessage implements
		KryoSerializable {

	private String schemaid;
	private String schema;

	public SchemaMetaMessage() {

	}

	public SchemaMetaMessage(String tenantid, String schemaId, String schema) {
		this.tenantid = tenantid;
		this.schemaid = schemaId;
		this.schema = schema;
		this.type = MessageType.ADD_SCHEMA;
		this.processorid = schemaId;
	}

	/**
	 * Json format 
	 * tenantid schemaid schema
	 * @param schemaStr
	 */
	public SchemaMetaMessage(byte[] schemaByte) {
		try {
			String schemaStr = new String(schemaByte);
			ObjectNode object = JSONUtility.fromObjectString(schemaStr);
			tenantid = object.get(MessageFields.TENANTID).asText();
			type = MessageType.ADD_SCHEMA;
			schemaid = object.get(MessageFields.SCHEMAID).asText();
			ArrayNode node = (ArrayNode) object.get(MessageFields.SCHEMA);
			schema = node.toString();
			System.out.println("schema in metadata is " + schema);
		} catch (CPSException e) {
			e.printStackTrace();
		}
	}

	public String getSchema() {
		return schema;
	}

	public String getSchemaId() {
		return schemaid;
	}

	@Override
	public void write(Kryo kryo, Output output) {
		output.writeString(tenantid);
		output.writeString(schemaid);
		output.writeString(schema);
		kryo.writeObject(output, MessageType.ADD_SCHEMA);
	}

	@Override
	public void read(Kryo kryo, Input input) {
		tenantid = input.readString();
		schemaid = input.readString();
		schema = input.readString();
		type = kryo.readObject(input, MessageType.class);
	}

	public String getJsonString() {
		Gson gson = new Gson();
		String json = gson.toJson(this);
		try {
			ObjectNode objectNode = JSONUtility.fromObjectString(json);
			objectNode.remove(MessageFields.TYPE);
			objectNode.put(MessageFields.TYPE, "schema");
			objectNode.set(
					MessageFields.SCHEMA,
					(ArrayNode) JSONUtility.fromString(objectNode.get(
							MessageFields.SCHEMA).asText()));
			return objectNode.toString();
		} catch (CPSException e) {
			e.printStackTrace();
		}
		return null;
	}

	public static void main(String[] args) {
		String tenantid = "tenantid";
		String schemaid = "123";
		String schema = "[\"name\":phase, \"dataType\":int]";
		SchemaMetaMessage metaMessage = new SchemaMetaMessage(tenantid, schemaid,
				schema);
		SchemaMetaMessage newMeta = new SchemaMetaMessage(metaMessage
				.getJsonString().getBytes());
		System.out.println(newMeta.getUserId());
	}

	@Override
	public void updateProcessorMap(
			HashMap<String, Set<String>> parent2Successors) {
		// nothing to do
	}
}
